package com.navinfo.tripanalysis.service.impl;

import com.mongodb.client.MongoCollection;
import com.mongodb.client.model.Filters;
import com.mongodb.client.result.DeleteResult;
import com.mongodb.spark.MongoConnector;
import com.mongodb.spark.MongoSpark;
import com.mongodb.spark.config.WriteConfig;
import com.navinfo.tripanalysis.common.arithmetic.common.OuterEventData;
import com.navinfo.tripanalysis.common.arithmetic.convert.EventDataMongoConvert;
import com.navinfo.tripanalysis.common.util.DateUtils;
import com.navinfo.tripanalysis.service.SaveEventService;
import com.navinfo.tripanalysis.util.MongoUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.SparkSession;
import org.bson.Document;
import org.bson.conversions.Bson;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Serializable;
import scala.Tuple2;

import java.util.*;

/**
 * 保存事件数据到Mongo中
 *
 * @author 沈东生
 */
public class SaveEventServiceMongoImpl implements SaveEventService, Serializable {
    private static final Logger logger = LoggerFactory.getLogger(SaveEventServiceMongoImpl.class);
    private String collectionNamePrefix;
    private Properties config;

    public Properties getConfig() {
        return config;
    }

    public void setConfig(Properties config) {
        this.config = config;
    }

    public String getCollectionNamePrefix() {
        return collectionNamePrefix;
    }

    public void setCollectionNamePrefix(String collectionNamePrefix) {
        this.collectionNamePrefix = collectionNamePrefix;
    }

    @Override
    public void save(SparkSession spark, JavaSparkContext jsc, JavaPairRDD<Long, List<OuterEventData>> eventRDD, long d) {
        long start = System.currentTimeMillis();
        logger.error("进行事件统计信息处理，保存mongo开始...");

        String dayStr = DateUtils.format(new Date(d), DateUtils.DateFormat.YYYYMMDD);
        String collectionName = getCollectionNamePrefix() + MongoUtils.getWeekTableName(d);

        Map<String, String> map = new HashMap<String, String>();
        map.put("collection", collectionName);

        WriteConfig writeConfig = WriteConfig.create(jsc).withOptions(map);

        //按天删除旧数据
        MongoConnector.apply(jsc.sc()).withCollectionDo(writeConfig, Document.class, (Function<MongoCollection<Document>, Object>) collection -> {
            Bson dayFilter = Filters.eq("day", dayStr);
            DeleteResult result = collection.deleteMany(dayFilter);
            logger.info("delete:{}", result);
            return result;
        });
        logger.error("进行事件统计信息处理，删除mongo旧数据耗时{}ms", System.currentTimeMillis()-start);

        JavaRDD<Document> documents = eventRDD
            .flatMap((FlatMapFunction<Tuple2<Long, List<OuterEventData>>, OuterEventData>) e ->  e._2().iterator())
            .mapPartitions((FlatMapFunction<Iterator<OuterEventData>, Document>) interDataIterator ->
                new Iterator<Document>() {
                    @Override
                    public boolean hasNext() {
                        return interDataIterator.hasNext();
                    }
                    @Override
                    public Document next() {
                        return EventDataMongoConvert.toDocument( interDataIterator.next());
                    }
                });

        MongoSpark.save(documents, writeConfig);
        logger.error("进行事件统计信息处理，保存mongo耗时{}ms",  System.currentTimeMillis()-start);
    }
}
